package com.zlx.rulemk.marketing.functions;

import com.alibaba.fastjson.JSON;
import com.zlx.rulemk.marketing.beans.*;
import com.zlx.rulemk.marketing.utils.FlinkStateDescUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;


/**
 * @Author: zhanglingxing
 * @CreateTime: 2022-07-10
 * @Description: 一条日志可能对应多条规则，进行数据扩展
 * @Version: 1.0
 *  这个类的目的就是动态分组
 *      一条日志可能对应多条规则，进行数据扩展
 *  这个版本主要是将 CDCLogBean 对象传递下去
 */
@Slf4j
public class DynamicKeyByReplicationFunctionV2 extends BroadcastProcessFunction<EventBean, CDCLogBean, DynamicKeyedBean> {

    /**
     * 这个地方设计的很巧妙
     * 我们后面数据的分组是
     * 并行度随机数+规则ID
     * 这样分组的目的是为了提高系统的并行处理能力
     * 为了缓存发送到下游的规则 用map暂存一下
     *  key：随机并行度+":"+ruleName
     */
    private Map<String,CDCLogBean> cdcLogBeanMaps =new ConcurrentHashMap<String,CDCLogBean>();

    private Integer flink_parallelism;

    public DynamicKeyByReplicationFunctionV2(Integer flink_parallelism){
        this.flink_parallelism = flink_parallelism;
    }

    /*
        数据驱动
      动态keyby数据复制
     */
    @Override
    public void processElement(EventBean eventBean, ReadOnlyContext ctx, Collector<DynamicKeyedBean> out) throws Exception {

        ReadOnlyBroadcastState<String, List<DynamicKeyRuleStateBean>> broadcastState = ctx.getBroadcastState(FlinkStateDescUtils.ruleStateDesc1);

        List<DynamicKeyRuleStateBean> dynamicKeyRuleStateBeans = broadcastState.get(eventBean.getEventId());
        if(dynamicKeyRuleStateBeans == null || dynamicKeyRuleStateBeans.size() ==0) {
            log.warn("事件ID：{},在系统没有对应的规则，不处理该事件，也不缓存",eventBean.getEventId());
            return ;
        }

        for (DynamicKeyRuleStateBean dynamicKeyRuleStateBean : dynamicKeyRuleStateBeans ) {

            String keyByFields = dynamicKeyRuleStateBean.getKeyByFields();

            // 规则要求的触发事件
            String ruleTriggerEventId = dynamicKeyRuleStateBean.getRuleTriggerEventId();

            String ruleName = dynamicKeyRuleStateBean.getRuleName();

            // 只有规则触发事件等于触发事件的时候 我们才会动态分组
            if(ruleTriggerEventId.equals(eventBean.getEventId())){

                //获取分组key对应的具体value的值
                StringBuilder sb = new StringBuilder();
                String[] fieldNames = keyByFields.split(",");
                // 拼装keyByFields中指定的每一个字段的值
                for (String fieldName : fieldNames) {
                    Class<?> beanClass = Class.forName("com.zlx.rulemk.marketing.beans.EventBean");
                    Field declaredField = beanClass.getDeclaredField(fieldName);
                    declaredField.setAccessible(true);
                    // 从eventbean中取分组字段的值
                    String fieldValue = (String)declaredField.get(eventBean);
                    sb.append(fieldValue).append(",");
                }
                // 获取具体的分组字段的
                String keyByValue = sb.substring(0, sb.length() - 1);
                eventBean.setKeyByValue(keyByValue);

                // 后面数据分组的key
                String key = (Math.abs(keyByValue.hashCode()) % flink_parallelism)+":"+eventBean.getEventId();
                CDCLogBean cdcLogBean = cdcLogBeanMaps.get(key);
                if(cdcLogBean != null){ //只有第一条数据才携带规则
                    cdcLogBeanMaps.remove(key);
                }
                DynamicKeyedBean dynamicKeyedBean = new DynamicKeyedBean(keyByValue, keyByFields, eventBean,ruleName, key,cdcLogBean);

                // 完成了 动态数据扩展 一条日志 可能对应多条规则 进行了数据扩张
                out.collect(dynamicKeyedBean);
            }
        }
    }

    /**
     * 插入mysql规则数据进行触发下面的方法
     */
    @Override
    public void processBroadcastElement(CDCLogBean cdcLogBean, Context ctx, Collector<DynamicKeyedBean> out) throws Exception {

        // 每个事件有多条规则 事件ID 对应多条规则
        BroadcastState<String, List<DynamicKeyRuleStateBean>> broadcastState = ctx.getBroadcastState(FlinkStateDescUtils.ruleStateDesc1);

        // 规则平台上操作的规则的数据
        List<RuleTableRecord> ruleTableRecords = cdcLogBean.getData();
        if(ruleTableRecords == null ) {
            log.warn("规则触发。接受到的规则数据为null，不进行处理");
            return;
        }

        // 规则平台上的操作的类型
        String operationType = cdcLogBean.getType();

        // 根据不同的操作类型，去操作state中的规则数据
        for (RuleTableRecord ruleTableRecord : ruleTableRecords) {

            String rule_condition_json = ruleTableRecord.getRule_condition_json();
            MarketingRule marketingRule = JSON.parseObject(rule_condition_json, MarketingRule.class);
            String keyByFields = marketingRule.getKeyByFields();
            // 事件ID 我们的规则都是针对某一个事件配置的规则
            String triggerEventId = marketingRule.getTriggerEventCondition().getEventId();
            // 这条规则的唯一ID
            String rule_name = ruleTableRecord.getRule_name();

            List<DynamicKeyRuleStateBean> dynamicKeyRuleStateBeans = broadcastState.get(triggerEventId);
            if(dynamicKeyRuleStateBeans == null){
                dynamicKeyRuleStateBeans = new ArrayList<>();
            }
            // 新增规则
            if ("INSERT".equals(operationType) || "UPDATE".equals(operationType) && "1".equals(ruleTableRecord.getRule_status())) {

                if("INSERT".equals(operationType) && dynamicKeyRuleStateBeans.size() == 0){ //第一次插入
                    for (Integer i = 0; i < flink_parallelism; i++) {
                        cdcLogBeanMaps.put(i+":"+triggerEventId,cdcLogBean);
                    }
                    DynamicKeyRuleStateBean dynamicKeyRuleStateBean = new DynamicKeyRuleStateBean(triggerEventId,rule_name,keyByFields);
                    dynamicKeyRuleStateBeans.add(dynamicKeyRuleStateBean);
                    //将规则新增到state中
                    broadcastState.put(triggerEventId,dynamicKeyRuleStateBeans);
                    log.debug(getRuntimeContext().getIndexOfThisSubtask()+">>> :新增规则成功，规则对应的eventId：{},简化规则:{}",triggerEventId,dynamicKeyRuleStateBeans);
                }else {
                    //都是更新的情况  更新规则
                    for (DynamicKeyRuleStateBean stateBean : dynamicKeyRuleStateBeans) {
                        if(stateBean.getRuleName().equals(rule_name)){ // 只 更新对应的规则 即可
                            stateBean.setKeyByFields(keyByFields);
                            stateBean.setRuleTriggerEventId(triggerEventId);
                            for (Integer i = 0; i < flink_parallelism; i++) {
                                cdcLogBeanMaps.put(i+":"+triggerEventId,cdcLogBean);
                            }
                            break;
                        }
                    }
                    //将规则新增到state中
                    broadcastState.put(triggerEventId,dynamicKeyRuleStateBeans);
                    log.debug(getRuntimeContext().getIndexOfThisSubtask()+">>> :更新操作规则成功，规则对应的eventId：{},简化规则:{}",triggerEventId,dynamicKeyRuleStateBeans);
                }
            }

            if ("DELETE".equals(operationType) || ("UPDATE".equals(operationType) && "0".equals(ruleTableRecord.getRule_status()))) {
                // 如果该keybyfields对应的ruleList已经被删空，则直接从广播state中将这个keybyFields条目都删除
                if(dynamicKeyRuleStateBeans.size() == 0){
                    return ;
                }else if(dynamicKeyRuleStateBeans.size() == 1){
                    //移除整个规则
                    broadcastState.remove(triggerEventId);

                    // 同时移除
                    for (Integer i = 0; i < flink_parallelism; i++) {
                        cdcLogBeanMaps.remove(i+":"+triggerEventId);
                    }
                    log.debug(getRuntimeContext().getIndexOfThisSubtask()+">>> :规则删除成功，删除事件ID：{}对应的所有规则",triggerEventId);
                }else{
                    // 删除具体的一条规则
                    Iterator<DynamicKeyRuleStateBean> iterator = dynamicKeyRuleStateBeans.iterator();
                    while (iterator.hasNext()){
                        DynamicKeyRuleStateBean stateBean = iterator.next();
                        if(stateBean.getRuleName().equals(rule_name)){
                            //删除该规则
                            iterator.remove();
                            log.debug(getRuntimeContext().getIndexOfThisSubtask()+">>> :规则删除成功，删除事件ID：{}，对应的具体简化规则:{}",triggerEventId,stateBean);
                            break;//跳出while循环
                        }
                    }

                    // 同时移除
                    for (Integer i = 0; i < flink_parallelism; i++) {
                        cdcLogBeanMaps.remove(i+":"+triggerEventId);
                    }
                }


            }

        }

    }

}
